package com.rr.learningdemo.flink;

import org.apache.commons.collections.IteratorUtils;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.streaming.util.serialization.SimpleStringSchema;
import org.apache.flink.util.Collector;

import java.util.Properties;

/**
 * @author zhuangcanrun
 * @Date 2021/08/10/19:34
 * @description
 **/

public class KafkaSavePointTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        Properties properties = new Properties();
        //配置好kafka的地址，记得开端口，虚拟机图省事直接把防火墙关了
        properties.setProperty("bootstrap.servers", "172.16.5.33:9092");
        //要监听的topic是test，消费一些简单的string消息
        FlinkKafkaConsumer<String> consumer = new FlinkKafkaConsumer<>("test", new SimpleStringSchema(), properties);
        //从最早开始消费
        consumer.setStartFromEarliest();
        DataStream<String> stream = env.addSource(consumer);
        stream.print();
        stream.windowAll(TumblingProcessingTimeWindows.of(Time.minutes(2)))
                .process(new ProcessAllWindowFunction<String, Object, TimeWindow>() {
                    @Override
                    public void process(Context context, Iterable<String> iterable, Collector<Object> collector) throws Exception {
                        System.out.println("=======" + IteratorUtils.toArray(iterable.iterator()).length + "=========");
                    }
                });
        //stream.map();
        env.execute();

    }
}
